package ReduceJoin;

import jdk.nashorn.internal.ir.CallNode;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

public class RJReducer extends Reducer<Text,TableBean,TableBean, NullWritable> {


    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        /*
         *01 key值唯一
         * {01,{TableBean1,TableBean2,TableBean3,TableBean4}}
         * TableBean1={order_id:201801,p_id:01,amount:1,name:"",flag="0"}
         * TableBean2={order_id:"",p_id:01,amount:0,name:苹果,flag="1"}
         */
        List<TableBean> orderBean=  new ArrayList<>();
        TableBean pdBean= new TableBean();
        //遍历TableBean
        for (TableBean value: values) {
            TableBean order= new TableBean();
            if(value.getFlag().equals("0")){//如果是order表
                try {
                    //将遍历order表中的TableBean(value)拷贝到一个TableBean（order）
                    BeanUtils.copyProperties(order,value);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
                //如果是order表放入到list集合里（）
                orderBean.add(order);
            }else{//如果是pd表
                try {
                    BeanUtils.copyProperties(pdBean,value);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }

            }


        }
        //设置产品的名字
        //遍历list集合
        for ( TableBean bean:orderBean) {
            bean.setName(pdBean.getName());
            //写出
            context.write(bean,NullWritable.get());
        }
    }
}
